package com.unknownman.block.transaction.subscribe;

import com.unknownman.block.component.MessageSolverChooser;
import com.unknownman.block.constants.ChainContractConstants;
import com.unknownman.block.constants.LogConstants;
import com.unknownman.block.dto.ChainContractComposition;
import com.unknownman.block.entity.TContractConfig;
import com.unknownman.block.entity.TContractLogs;
import com.unknownman.block.service.TChainConfigService;
import com.unknownman.block.service.TContractConfigService;
import com.unknownman.block.service.TContractLogsService;
import lombok.extern.slf4j.Slf4j;
import org.cloud.utils.CollectionUtil;
import org.cloud.utils.SpringContextUtil;
import org.springframework.util.Assert;
import org.web3j.protocol.Web3j;
import org.web3j.protocol.core.DefaultBlockParameter;
import org.web3j.protocol.core.methods.request.EthFilter;
import org.web3j.protocol.core.methods.response.EthBlock;
import org.web3j.protocol.core.methods.response.EthLog;
import org.web3j.protocol.core.methods.response.Log;
import org.web3j.protocol.core.methods.response.TransactionReceipt;
import org.web3j.protocol.http.HttpService;

import java.math.BigInteger;
import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;

@Slf4j
public class EthTranFilter4 extends BaseTranFilter {

    private boolean stop = false;
    private long checkInterval = 2000;

    private final MessageSolverChooser messageSolverChooser;

    private final TChainConfigService chainConfigService;
    private final TContractLogsService contractLogsService;
    private final TContractConfigService contractConfigService;

    private final ChainContractComposition chainContractComposition;

    private final Map<String, TContractConfig> contractConfigMap;

    private final List<String> rpcList;

    public EthTranFilter4(ChainContractComposition chainContractComposition, List<String> rpcList) {
        super();
        chainConfigService = SpringContextUtil.getBean(TChainConfigService.class);
        contractLogsService = SpringContextUtil.getBean(TContractLogsService.class);
        messageSolverChooser = SpringContextUtil.getBean(MessageSolverChooser.class);
        contractConfigService = SpringContextUtil.getBean(TContractConfigService.class);
        this.chainContractComposition = chainContractComposition;
        //查出启用的合约地址
        contractConfigMap = this.chainContractComposition.getContractList().stream()
            .filter(cont -> cont.getContractStatus() == ChainContractConstants.ContractStatusConstants.NORMAL.getStatus())
            .collect(Collectors.toMap((contractConfig) -> contractConfig.getContractAddress().toUpperCase(), Function.identity()));
        this.rpcList = rpcList;
        this.createWe3j();
        try {
            BigInteger chainId = web3j.ethChainId().sendAsync().get().getChainId();
            Assert.isTrue(chainId.intValue() == chainContractComposition.getChainId(), "RPC连接配置错误，chainId不匹配");
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException(e.getMessage());
        }

    }

    /**
     * 添加监听
     */
    public boolean addFilter(BigInteger startBlock) {
        log.info("addFilter: 开始获取事件交易, [startBlock: {}]", startBlock);
        lastBlock = startBlock;
        stop = false;
        new Thread(() -> {
            long nextCheck = 0;
            while (!(Thread.interrupted() || stop)) {
                if (nextCheck <= System.currentTimeMillis()) {
                    nextCheck = System.currentTimeMillis() + checkInterval;
                    try {
                        runScan();
                    } catch (Exception e) {
                        stop = true;
                        log.error("run scan error::[{}]", e.getMessage());
                        e.printStackTrace();
                    }
                } else {
                    try {
                        Thread.sleep(Math.max(nextCheck - System.currentTimeMillis(), 500));
                    } catch (InterruptedException ex) {
                        log.error("addFilter3: 扫描区块获取事件异常, [lastBlock: {}, e: {}]", lastBlock, ex.getMessage());
                        break;
                    }
                }
            }
            log.warn("addFilter3: 结束扫描区块获取事件, [lastBlock: {}]", lastBlock);
        }).start();
        return true;
    }


    private void runScan() throws Exception {
        long currentBlockNumber = lastBlock.longValue();
        long startBlock = currentBlockNumber + 1;
        //事件的确认区块
        BigInteger confirmBlock = chainContractComposition.getConfirmBlock() == null ? BigInteger.valueOf(10) : BigInteger.valueOf(chainContractComposition.getConfirmBlock());
        //当前最新区块暂时不处理
        BigInteger latest = web3j.ethBlockNumber().sendAsync().get().getBlockNumber().subtract(confirmBlock);
        if (latest.longValue() <= currentBlockNumber) {
            return;
        }
        //最大查询5000区块
        BigInteger endBlock = BigInteger.valueOf(latest.longValue() - startBlock > 999 ? startBlock + 999 : latest.longValue());
        List<String> contractAddressList = new ArrayList<>(contractConfigMap.keySet());
        EthFilter ethFilter = new EthFilter(DefaultBlockParameter.valueOf(BigInteger.valueOf(startBlock)),
            DefaultBlockParameter.valueOf(endBlock),
            contractAddressList);
        EthLog ethLog = web3j.ethGetLogs(ethFilter).send();
        Assert.isTrue(!ethLog.hasError(),
            "获取事件异常-" + (ethLog.getError() != null && ethLog.getError().getMessage() != null ? ethLog.getError().getMessage() : ""));
        List<EthLog.LogResult> logs = ethLog.getLogs();
        if (CollectionUtil.single().isNotEmpty(logs)) {
            subscribeCallBack(logs);
        }
        lastBlock = endBlock;
        updateLastBlock(lastBlock, true);
    }


    @Override
    public void subscribeCallBack(TransactionReceipt transactionReceipt) throws Exception {
        throw new RuntimeException("当前方法不允许调用");
    }

    public void subscribeCallBack(List<EthLog.LogResult> logs) throws Exception {
        if (CollectionUtil.single().isEmpty(logs)) {
            return;
        }
        HashMap<String, EthBlock.Block> blockMap = new HashMap<>();
        ArrayList<TContractLogs> saveLogList = new ArrayList<>();
        for (EthLog.LogResult logResult : logs) {
            Log log = (Log) logResult.get();
            String contractAddress = log.getAddress();
            //判断是不是监听的日志
            TContractConfig contractConfig = contractConfigMap.get(contractAddress.toUpperCase());
            if (contractConfig == null) {
                continue;
            }
            String transactionHash = log.getTransactionHash();
            Long logIndex = log.getLogIndex().longValue();
            if (!contractLogsService.existsByHashAndIndex(transactionHash, logIndex)) {
                EthBlock.Block block = blockMap.get(log.getBlockHash());
                if (block == null) {
                    block = web3j.ethGetBlockByHash(log.getBlockHash(), false).send().getBlock();
                    blockMap.put(log.getBlockHash(), block);
                }
                List<String> topics = log.getTopics();
                String data = log.getData();
                data = data.equals("0x") ? "0x0" : data;
                TContractLogs contractLogs = new TContractLogs();
                contractLogs.setContractConfigId(contractConfig.getContractConfigId());
                contractLogs.setContractAddress(contractAddress);
                contractLogs.setTransHash(transactionHash);
                contractLogs.setIndexForBlock(logIndex);
                contractLogs.setBlockNumber(block.getNumber().longValue());
                contractLogs.setBlockTimestamp(new Date(block.getTimestamp().longValue() * 1000));
                contractLogs.setRpcAddress(chainContractComposition.getRpcAddress());

                contractLogs.setFirstTopic(topics.get(0));
                if (topics.size() >= 2) {
                    contractLogs.setSecondTopic(topics.get(1));
                }
                if (topics.size() >= 3) {
                    contractLogs.setThirdTopic(topics.get(2));
                }
                if (topics.size() >= 4) {
                    contractLogs.setFourthTopic(topics.get(3));
                }
                contractLogs.setLogData(data);
                contractLogs.setStatus(LogConstants.ContractLogStatus.UNTREATED.getStatus());
                saveLogList.add(contractLogs);
                //消息发送
                //根据合约分发消息
//                MessageSolver messageSolver = messageSolverChooser.choose(logChain.getAddress().toUpperCase());
//                if (messageSolver != null) {
//                    messageSolver.solve(logChain);
//                }
            }
        }
        if (CollectionUtil.single().isNotEmpty(saveLogList)) {
            boolean b = contractLogsService.saveBatch(saveLogList);
            Assert.isTrue(b, "subscribeCallBack-" + chainContractComposition.getChainSymbol() + "：log保存失败");
        }

    }

    public void updateLastBlock(BigInteger block, boolean success) {
        chainConfigService.updateLastBlock(chainContractComposition.getChainConfigId(), block.longValue(), success);
    }

    /**
     * 订阅是否关闭
     *
     * @return
     */
    @Override
    public boolean isDisposed() {
        return stop;
    }

    /**
     * 关闭订阅
     */
    @Override
    public void stopSubscribe() {
        stop = true;
    }

    private void createWe3j() {
        chainContractComposition.setRpcAddress(this.rpcList.get(new Random().nextInt(this.rpcList.size())));
        try {
            web3j = Web3j.build(new HttpService(chainContractComposition.getRpcAddress()));
        } catch (Exception e) {
            log.info("初始化w3[{}]异常，请检查:{}", chainContractComposition.getRpcAddress(), e.getMessage());
            return;
        }
        this.rpcAddress = chainContractComposition.getRpcAddress();
        log.info("初始化w3[{}]成功！", chainContractComposition.getRpcAddress());
    }
}
